package com.xiguthings.xiniu.iot.post.kafka;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xiguthings.iot.kafka.KafkaClienAstract;
import com.xiguthings.iot.kafka.consumer.base.KafkaConsumerConfig;
import com.xiguthings.iot.kafka.producer.base.KafkaProducerConfig;
import com.xiguthings.xiniu.iot.common.PostData;
import com.xiguthings.xiniu.iot.post.controller.DataPostContrller;

@Component
public class KafkaClient extends KafkaClienAstract {
	private final Logger LOGGER = LoggerFactory.getLogger(KafkaClient.class);

	// public static int count = 0;

	@Autowired
	private DataPostContrller dataPostContrller;

	private com.xiguthings.iot.kafka.producer.DataProducer dataProducer;

	@Value("${kafka_consumer_bootstrap_servers}")
	public String consumerServers;
	@Value("${kafka.enable.auto.commit}")
	public boolean enableAutoCommit;
	@Value("${kafka.session.timeout.ms}")
	public String sessionTimeout;
	@Value("${kafka.auto.commit.interval.ms}")
	public String autoCommitInterval;
	@Value("${kafka.auto.offset.reset}")
	public String autoOffsetReset;
	@Value("${kafka.post-data.topic}")
	private String topic;
	@Value("${kafka.consumer.group-id}")
	private String groupId;

	@Value("${kafka.producer.acks}")
	public String acks;
	@Value("${kafka_producer_bootstrap_servers}")
	public String producerServers;
	@Value("${kafka.producer.retries}")
	public int retries;
	@Value("${kafka.producer.buffer.memory}")
	public long bufferMemorry;
	@Value("${kafka.producer.batch.size}")
	public int bacthSize;
	@Value("${kafka.producer.linger.ms}")
	public long lingerMs;

	@PostConstruct
	private void action() {
		// 创建生产者
		KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig(acks, producerServers, retries, bufferMemorry,
				bacthSize, lingerMs);
		this.dataProducer = new com.xiguthings.iot.kafka.producer.DataProducer(kafkaProducerConfig);

		// 开启消费线程
		Thread consumerThread = new Thread(new Runnable() {
			@Override
			public void run() {
				KafkaConsumerConfig kafkaConsumerConfig = new KafkaConsumerConfig(consumerServers, enableAutoCommit,
						sessionTimeout, autoCommitInterval, autoOffsetReset);
				// 创建消费者
				com.xiguthings.iot.kafka.consumer.Consumer consumer = new com.xiguthings.iot.kafka.consumer.Consumer(
						topic, groupId, kafkaConsumerConfig);
				consumMessage(consumer, 100);
			}
		});
		consumerThread.start();

	}

	@Override
	public void handlerMessage(com.xiguthings.iot.kafka.consumer.Consumer consumer, String dataStr) {
		// count++;
		LOGGER.info("收到数据：{}", dataStr);
		PostData data = null;
		try {
			data = JSON.parseObject(dataStr, PostData.class);
		} catch (Exception e) {
			LOGGER.info("非法数据：{}", dataStr);
			return;
		}
		dataPostContrller.receiveData(data);
	}

	/**
	 * 生产消息
	 *
	 * @param data
	 */
	public void produceData(PostData data) {
		super.produceData(dataProducer, JSONObject.toJSONString(data), topic, null);
	}
}
